package com.zhang.third.day03;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @title: 副函数初体验
 * @author: zhang
 * @date: 2022/4/6 09:57
 */
public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .fromElements(1, 2, 3, 4, 5, 6, 7, 8)
                .keyBy(r -> r % 3)
                .map(new RichMapFunction<Integer, String>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        System.out.println("并行子任务索引是:" + getRuntimeContext().getIndexOfThisSubtask() + "" +
                                "的生命周期开始");
                    }

                    @Override
                    public String map(Integer value) throws Exception {
                        return "并行子任务索引是：" + getRuntimeContext().getIndexOfThisSubtask() + "" +
                                "，输出的数据是：" + value;
                    }

                    @Override
                    public void close() throws Exception {
                        System.out.println("并行子任务索引是:" + getRuntimeContext().getIndexOfThisSubtask() + "" +
                                "的生命周期结束");
                    }
                }).setParallelism(1)
                .print().setParallelism(2);

        env.execute();
    }
}
